package spark.hope

//import org.apache.spark.mllib.linalg.distributed.RowMatrix
import org.apache.spark.mllib.linalg.distributed.RowMatrix
import org.apache.spark.mllib.linalg.{Matrix, Vector, Vectors}
import org.apache.spark.mllib.stat.MultivariateStatisticalSummary
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Row, SQLContext, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}

object CovarianceExample {
  def main(args: Array[String]) {
    val spark = SparkSession.builder().appName("CovarianceExample").master("local[*]").getOrCreate()
//    val conf = new SparkConf().setAppName("CovarianceExample").setMaster("local[*]")
//    val sc = new SparkContext(conf)
//    val sqlContext = new SQLContext(sc)

    //输入数据
    val data = Array(
      Vectors.dense(4.0, 2.0, 3.0),
      Vectors.dense(5.0, 6.0, 1.0),
      Vectors.dense(2.0, 4.0, 7.0),
      Vectors.dense(3.0, 6.0, 5.0)
    )

    import spark.implicits._
    // Array[Vector]转换成DataFrame
    val df = spark.createDataFrame(data.map(Tuple1.apply)).toDF("features")
    df.show()

    // DataFrame转换成RDD
    val rddData: RDD[Vector] =df.select("features").map { case Row(v: Vector) => v}.rdd

    // RDD转换成RowMatrix
    val mat: RowMatrix = new RowMatrix(rddData)

    // 统计
    val stasticSummary: MultivariateStatisticalSummary =mat.computeColumnSummaryStatistics()

    // 均值
    println(stasticSummary.mean)
    // 结果：3.5,4.5,4.0

    // 方差
    println(stasticSummary.variance)
    // 结果：1.6666666666666667,3.6666666666666665,6.666666666666667


    // 协方差
    val covariance: Matrix = mat.computeCovariance()
    println(covariance)
    // 结果：
    //  cov(dim1,dim1) cov(dim1,dim2) cov(dim1,dim3)
    //  cov(dim2,dim1) cov(dim2,dim2) cov(dim2,dim3)
    //  cov(dim3,dim1) cov(dim3,dim2) cov(dim3,dim3)
    //  1.6666666666666679   0.3333333333333357   -3.3333333333333304
    //  0.3333333333333357   3.666666666666668    -0.6666666666666679
    //  -3.3333333333333304  -0.6666666666666679  6.666666666666668
    // 结果分析：以cov(dim1,dim2)为例
    //  dim1均值：3.5  dim2均值：4.5
    //  val cov(dim2,dim3)=((4.0-3.5)*(2.0-4.5)+(5.0-3.5)*(6.0-4.5)+(2.0-3.5)*(4.0-4.5)+(3.0-3.5)*(6.0-4.5))/(4-1)
    //  cov(dim2,dim3)=0.3333333333333333
  }

}